package xyz.xiezc.mzix.schedule.service;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.core.util.StrUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import xyz.xiezc.mzix.common.api.Schedule;
import xyz.xiezc.mzix.common.dto.PageDTO;
import xyz.xiezc.mzix.schedule.dao.PageRepository;
import xyz.xiezc.mzix.schedule.dao.SessionRepository;
import xyz.xiezc.mzix.schedule.entity.PageDO;
import xyz.xiezc.mzix.schedule.entity.SessionDO;
import xyz.xiezc.mzix.schedule.feign.DownloadService;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

@Component
public class ScheduleService implements Schedule {

    @Autowired
    private PageRepository pageRepository;

    @Resource
    private DownloadService downloadService;

    @Autowired
    private ReactiveRedisTemplate reactiveRedisTemplate;


    private ReactiveListOperations<String, PageDO> reactiveListOperations;

    @PostConstruct
    public void init() {
        reactiveListOperations = reactiveRedisTemplate.opsForList();
    }


    // @Scheduled(cron = "0/5 * * * * ?")
    public void execute() {
        String runCountStr = this.getSession("config", "runCount");
        String expireTimeStr = this.getSession("config", "expireTime");
        Long runCount = Convert.toLong(runCountStr, 1000L);
        Long expireTimeS = Convert.toLong(expireTimeStr, 30L);

        Mono<Long> wait = reactiveListOperations.size("wait");
        Mono<Long> running = reactiveListOperations.size("running");

        Mono.zip(wait, running).subscribe(it -> {
            Long t1 = it.getT1();
            if (t1 < runCount * 5) {
                long l = (runCount * 5000) - t1;
                List<PageDO> pageDOS = pageRepository.findByStatus(0, PageRequest.of(0, Convert.toInt(l)));

                pageDOS.stream().forEach(pageDO -> pageDO.setStatus(1));
                if (!pageDOS.isEmpty()) {
                    Mono<Long> wait1 = reactiveListOperations.leftPushAll("wait", pageDOS.toArray(new PageDO[0]));
                    wait1.subscribe(s -> {
                        pageRepository.saveAll(pageDOS);
                    });
                }
            }

            Long t2 = it.getT2();
            if (t2 < runCount) {
                Mono<PageDO> pageDOMono = reactiveListOperations.rightPopAndLeftPush("wait", "running");
                pageDOMono.subscribe(pageDO -> {
                    downloadService.download(BeanUtil.copyProperties(pageDO, PageDTO.class));
                });
            }
        });
    }


    @Override
    public void finish(PageDTO pageDTO) {
        PageDO pageDO = BeanUtil.copyProperties(pageDTO, PageDO.class);
        pageDO.setRequest(BeanUtil.beanToMap(pageDTO.getRequest()));
        pageDO.setResponse(BeanUtil.beanToMap(pageDTO.getResponse()));
        pageDO.setStatus(2);
        pageRepository.save(pageDO);
        ReactiveHashOperations reactiveHashOperations = reactiveRedisTemplate.opsForHash();
        reactiveHashOperations.remove("runningPage", pageDTO.getId());

        List<PageDTO> pageDTOS = pageDTO.nextPage();
        List<PageDO> collect = pageDTOS.parallelStream().map(p -> {
            PageDO pDO = BeanUtil.copyProperties(p, PageDO.class);
            pDO.setRequest(BeanUtil.beanToMap(p.getRequest()));
            pDO.setResponse(BeanUtil.beanToMap(p.getResponse()));
            pDO.setStatus(0);
            return pDO;
        }).collect(Collectors.toList());

        pageRepository.saveAll(collect);
    }


    @Override
    public String getSession(String code, String key) {
        ReactiveHashOperations<String, String, String> ckHash = reactiveRedisTemplate.opsForHash();
        Mono<String> value = ckHash.get(code, key);
        return value.block();
    }

    @Override
    public Boolean setSession(String code, String key, String value) {
        ReactiveHashOperations<String, String, String> ckHash = reactiveRedisTemplate.opsForHash();
        Mono<Boolean> put = ckHash.put(code, key, value);
        return put.toProcessor().block();
    }
}
